package org.example.sync;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.NavigableMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 生产者异步确认
 */
public class TopicsProducerConfirmAsyncTest {


    public static final String CONFIRM_EXCHANGE = "producer_confirm_individually";


    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            //声明广播类型的交换机
            channel.exchangeDeclare(CONFIRM_EXCHANGE, BuiltinExchangeType.DIRECT);
            //Publisher confirms are a RabbitMQ extension to the AMQP 0.9.1 protocol,
            // so they are not enabled by default.
            // Publisher confirms are enabled at the channel level with the confirmSelect method:
            channel.confirmSelect();
            //声明一个交换机
            long start = System.currentTimeMillis();
            ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            System.out.println("开始确认时间:" + start);
            for (int i = 0; i < 1000; i++) {
                String message = "消息" + i;
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish(CONFIRM_EXCHANGE, "", null, message.getBytes());
            }
            ConfirmCallback confirmCallback = new ConfirmCallback() {
                @Override
                public void handle(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple) {
                        //批量确认
                        NavigableMap<Long, String> headMap = outstandingConfirms.headMap(deliveryTag, true);
                        headMap.clear();
                    } else {
                        outstandingConfirms.remove(deliveryTag);
                    }
                }
            };
            channel.addConfirmListener(confirmCallback, new ConfirmCallback() {
                @Override
                public void handle(long deliveryTag, boolean multiple) throws IOException {
                    String body = outstandingConfirms.get(deliveryTag);
                    System.err.format(
                            "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
                            body, deliveryTag, multiple
                    );
                    confirmCallback.handle(deliveryTag, multiple);
                }
            });
            System.out.println("确认结束：" + (System.currentTimeMillis() - start) + "ms");
            // In the Java client, when we supply no parameters to queueDeclare() we
            // create a non-durable, exclusive, autodelete queue with a generated name:
            // AMQP.Queue.DeclareOk declareOk = channel.queueDeclare();
            // 获取一个非持久化，排他，自动删除的队列
            // String queueName = declareOk.getQueue();
            // channel.queueBind(queueName,PUB_EXCHANGE,"");
        }
    }

    private static String getRoutingKey(String[] args) {
        if (args.length < 1)
            return "anonymous.info";
        return args[0];
    }
}
